package com.allyes.flume.interceptor;

import static com.allyes.flume.source.TaildirSourceConfigurationConstants.*;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Arrays;
import java.util.Date;
import java.text.SimpleDateFormat;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import com.allyes.flume.interceptor.parser.LogParser;
import com.allyes.flume.interceptor.parser.LogParserFactory;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.conf.Configurable;
import org.apache.flume.interceptor.Interceptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.allyes.flume.source.Pair;
import com.allyes.flume.interceptor.rawlogdigger.RawlogDiggerConstants;
import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
import com.google.common.util.concurrent.ThreadFactoryBuilder;


public class RawlogInterceptor implements Interceptor {
  private static final Logger logger = LoggerFactory.getLogger(RawlogInterceptor.class);

  private class RawlogInterceptorMonitorRunnable implements Runnable {
    @Override
    public void run() {
      monitor.writeFileAndReset();
    }
  }

  static class MonitorItems {
    public long theBiggestTimestampInSec;
    public long theSmallestTimestampInSec;
    public long numRawlogs;

    public MonitorItems() {
      reset();
    }

    public void reset() {
      theBiggestTimestampInSec = 0l;
      theSmallestTimestampInSec = 0l;
      numRawlogs = 0l;
    }

    public void update(long timestampInSec) {
      if (timestampInSec > theBiggestTimestampInSec) {
        theBiggestTimestampInSec = timestampInSec;
      }
      if (0 == theSmallestTimestampInSec ||
          timestampInSec < theSmallestTimestampInSec) {
        theSmallestTimestampInSec = timestampInSec;
      }
      numRawlogs++;
    }

    public MonitorItems copy() {
      MonitorItems cpy = new MonitorItems();
      cpy.theBiggestTimestampInSec = this.theBiggestTimestampInSec;
      cpy.theSmallestTimestampInSec = this.theSmallestTimestampInSec;
      cpy.numRawlogs = this.numRawlogs;
      return cpy;
    }
  }

  static class Monitor implements Configurable {
    static public final String TXT_MONITOR = "monitor";
    static public final String TXT_MONITOR_PREFIX = TXT_MONITOR + ".";
    static public final String TXT_PRODUCT = "product";
    static public final String TXT_APP = "app";
    static public final String TXT_WRITE_DIR = "writeDir";
    static public final String TXT_WRITE_INTERVAL_IN_MINUTE = "writeIntervalInMinute";
    static public final String TXT_ALLOWED_MAX_DELAY_IN_MINUTE = "allowedMaxDelayInMinute";
    static public final String TXT_OK = "OK";
    static public final String TXT_CRITICAL = "CRITICAL";

    public static String getDateAndTime(Date date) {
      SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
      return dateFormat.format(date);
    }

    public static String getFileName(String product, String app, String service) {
      return "Status_" + product + "_" + app + "_" + service + "_update";
    }

    private String productName;
    private String appName;
    final private String serviceName = "flume";
    private int writeIntervalInMinute;
    private int allowedMaxDelayInMinute;
    private String fileToAppend;
    private MonitorItems monitorItems = new MonitorItems();

    @Override
    public void configure(Context context) {
      this.productName = context.getString(TXT_PRODUCT);
      this.appName = context.getString(TXT_APP);
      String writeDir = context.getString(TXT_WRITE_DIR);
      this.fileToAppend = writeDir + "/" + getFileName(productName, appName, serviceName);

      this.writeIntervalInMinute = context.getInteger(TXT_WRITE_INTERVAL_IN_MINUTE);
      this.allowedMaxDelayInMinute = context.getInteger(TXT_ALLOWED_MAX_DELAY_IN_MINUTE);
      Preconditions.checkState(allowedMaxDelayInMinute >= writeIntervalInMinute);
    }

    public void update(long timestampInSec) {
      synchronized(monitorItems) {
        monitorItems.update(timestampInSec);
      }
    }

    public MonitorItems getAndReset() {
      synchronized(monitorItems) {
        MonitorItems cpy = monitorItems.copy();
        monitorItems.reset();
        return cpy;
      }
    }

    public void writeFileAndReset() {
      try {
        logger.info("Begin to append monitor info to {}", fileToAppend);

        MonitorItems m = getAndReset();

        final Date now = new Date();
        final long timestampNow = now.getTime() / 1000;

        final String service_status;
        final String service_desc;
        if (0 == m.numRawlogs) {
          service_status = TXT_OK;
          service_desc = "No rawlog.";
        } else {
          if (timestampNow - m.theBiggestTimestampInSec <= (allowedMaxDelayInMinute * 60)) {
            service_status = TXT_OK;
          } else {
            service_status = TXT_CRITICAL;
          }
          service_desc = "c(" + timestampNow + ") " +
              "s(" + (m.theSmallestTimestampInSec - timestampNow) + ") " +
              "b(" + (m.theBiggestTimestampInSec  - timestampNow) + ") " +
              "a(" + (allowedMaxDelayInMinute * 60) + "s) " +
              "n(" + m.numRawlogs + " in " + (writeIntervalInMinute * 60) + "s)";
        }

        FileWriter fileWriter = new FileWriter(new File(fileToAppend), true);
        fileWriter.write(getDateAndTime(now) + "\t" +
            productName + "\t" +
            appName + "\t" +
            serviceName + "\t" +
            service_status + "\t" +
            service_desc + "\n");
        fileWriter.close();

      } catch (Throwable t) {
        logger.error("Uncaught exception when writing monitor info of rawlog interceptor", t);
      }
    }

  }

  private final LogParser logParser;
  private final HashSet<String> headersNeeded;
  private Monitor monitor;
  private ScheduledExecutorService monitorService;

  private RawlogInterceptor(
      LogParser logParser,
      Monitor monitor) {
    this.logParser = logParser;
    this.monitor = monitor;
    this.headersNeeded = new HashSet<String>(Arrays.asList(
        TXT_INODE,
        TXT_FILE_NAME,
        TXT_LINE_INDEX));
  }

  @Override
  public void initialize() {
    logger.info("Initializing Rawlog interceptor.");

    if (monitor != null) {
      monitorService = Executors.newSingleThreadScheduledExecutor(
          new ThreadFactoryBuilder().setNameFormat("rawlogInterceptorMonitor").build());
      monitorService.scheduleWithFixedDelay(
          new RawlogInterceptorMonitorRunnable(),
          monitor.writeIntervalInMinute,
          monitor.writeIntervalInMinute,
          TimeUnit.MINUTES);
    }
  }

  @Override
  public void close() {
    if (monitor != null) {
      try {
        logger.info("Begin to shutdown monitor service...");
        monitorService.shutdown();
        if (!monitorService.awaitTermination(2, TimeUnit.SECONDS)) {
          monitorService.shutdownNow();
        }
        logger.info("Succeeded to shutdown monitor service.");
      } catch (InterruptedException e) {
        logger.info("Interrupted while awaiting termination in RawlogInterceptor", e);
      }
    }
  }

  @Override
  public Event intercept(Event event) {
    // check headers first
    for (String h: this.headersNeeded) {
      if (!event.getHeaders().containsKey(h)) {
        logger.error("Event header not found in RawlogInterceptor: " + h);
        return null;
      }
    }

    final long inode = Long.valueOf(event.getHeaders().get(TXT_INODE));
    final String fileName = event.getHeaders().get(TXT_FILE_NAME);
    final int lineIndex = Integer.valueOf(event.getHeaders().get(TXT_LINE_INDEX));
    final String line = new String(event.getBody(), Charsets.UTF_8);

    try {
      Pair<Boolean, Event> parseResult = logParser.parseToEvent(inode, fileName, lineIndex, line);
      Event newEvent = parseResult.right;

      if (newEvent != null) {
        for (String oldKey: event.getHeaders().keySet()) {
          if (!newEvent.getHeaders().containsKey(oldKey)) {
            newEvent.getHeaders().put(oldKey, event.getHeaders().get(oldKey));
          }
        }
      }

      if (this.monitor != null && parseResult.left && newEvent != null) {
        String timestampStr = newEvent.getHeaders().get(RawlogDiggerConstants.TXT_TIMESTAMP);
        long timestamp = Long.valueOf(timestampStr) / 1000;
        this.monitor.update(timestamp);
      }

      return newEvent;

    } catch (IOException e) {
      logger.error("IOException when parse to RawLog: " + e + "; The event will be dropped!");
      return null;
    }
  }

  @Override
  public List<Event> intercept(List<Event> events) {
    List<Event> eventsRet = new LinkedList<Event>();

    for (Event e: events) {
      Event newEvent = intercept(e);
      if (newEvent != null) {
        eventsRet.add(newEvent);
      }
    }

    return eventsRet;
  }

  public static class Builder implements Interceptor.Builder {
    private InvalidLogHandler invalidLogHandler;
    private DynamicRawlogEventHeader dynamicRawlogEventHeader = new DynamicRawlogEventHeader();
    private Monitor monitor;
    private LogParser logParser;

    @Override
    public Interceptor build() {
      return new RawlogInterceptor(logParser, monitor);
    }

    @Override
    public void configure(Context context) {
      invalidLogHandler = InvalidLogHandlerBuild.build(new Context(
          context.getSubProperties(InvalidLogHandlerBuild.HANDLE_INVALID_LOG_PREFIX)));
      Preconditions.checkNotNull(invalidLogHandler);

      HashSet<String> dynamicHeaders = new HashSet<String>();

      // monitor
      monitor = null;
      Map<String, String> monitorProperties = context.getSubProperties(Monitor.TXT_MONITOR_PREFIX);
      if (monitorProperties != null && !monitorProperties.isEmpty()) {
        Context monitorContext = new Context(monitorProperties);
        monitor = new Monitor();
        monitor.configure(monitorContext);
        dynamicHeaders.add(RawlogDiggerConstants.TXT_TIMESTAMP);
      }

      // dynamic headers
      String dynamicHeadersStr = context.getString(TXT_DYNAMIC_HEADERS);
      if (dynamicHeadersStr != null) {
        for (String h: dynamicHeadersStr.split("\\s+")) {
          dynamicHeaders.add(h);
        }
      }
      dynamicRawlogEventHeader.initialize(dynamicHeaders);

      //set log parser
      logParser = LogParserFactory.getLogParser(context, invalidLogHandler, dynamicRawlogEventHeader);
    }

  }  // Builder ends.
}
